package com.meituan.hive.udf.generic;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by zhaoxiang on 2017/8/10.
 * 合并两个map
 * func(map1, map2)
 */
public class GenericUDFMapMerge extends GenericUDF{
    private transient MapObjectInspector mapOI;

    private final Map<Object,Object> retMap = new HashMap<>();

    @Override
    public ObjectInspector initialize(ObjectInspector[] arguments)
            throws UDFArgumentException {
        if (arguments.length < 2) {
            throw new UDFArgumentLengthException("The function MapMerge accepts at least 2 argument.");
        } else if (!(arguments[0] instanceof MapObjectInspector)) {
            throw new UDFArgumentTypeException(0, "\""
                    + ObjectInspector.Category.MAP.toString().toLowerCase()
                    + "\" is expected at function MapMerge, " + "but \""
                    + arguments[0].getTypeName() + "\" is found");
        }
        mapOI = (MapObjectInspector) arguments[0];

        return ObjectInspectorUtils.getStandardObjectInspector(mapOI);
    }

    @Override
    public Object evaluate(GenericUDF.DeferredObject[] arguments) throws HiveException {
        retMap.clear();
        //boolean
        for(DeferredObject deferredObject : arguments){
            Object mapObj = deferredObject.get();
            if(mapObj == null)
                continue;
            Map<?,?> mapVal = mapOI.getMap(mapObj);
            retMap.putAll(mapVal);
        }

        return retMap;
    }

    @Override
    public String getDisplayString(String[] children) {
        assert children.length == 1;
        return getStandardDisplayString("MapSumValueBykey", children);
    }
}
